#pragma once
#include "requestor.hpp"

namespace Client
{
    using namespace std;
    class TopicManger
    {
    public:
        using SubCallback = function<void(const string &key, const string &msg)>;
        using ptr = std::shared_ptr<TopicManger>;
        TopicManger(const Requestor::ptr &requestor) : _requestor(requestor) {}
        bool Create(const BaseConnection::ptr &conn, const string &key)
        {
            return CommRequest(conn, key, TopicOptype::TOPIC_CREATE);
        }
        bool Remove(const BaseConnection::ptr &conn, const string &key)
        {
            return CommRequest(conn, key, TopicOptype::TOPIC_REMOVE);
        }
        // 订阅功能--处理
        bool Subscribe(const BaseConnection::ptr &conn, const string &key, const SubCallback &cb)
        {
            AddSubscribe(key, cb);
            bool ret = CommRequest(conn, key, TopicOptype::TOPIC_SUBSCRIBE);
            if (ret == false)
            {
                DelSubscribe(key);
                return false;
            }
            return true;
        }
        bool Cancel(const BaseConnection::ptr &conn, const string &key)
        {
            DelSubscribe(key);
            return CommRequest(conn, key, TopicOptype::TOPIC_CANCEL);
        }
        bool Publish(const BaseConnection::ptr &conn, const string &key, const string &msg)
        {
            return CommRequest(conn, key, TopicOptype::TOPIC_PUBLISH, msg);
        }
        void OnPublish(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
        {
            auto type = msg->GetOptype();
            if (type != TopicOptype::TOPIC_PUBLISH)
            {
                ELog("收到了错误的类型的主题操作！");
                return;
            }
            // 取出消息主题名称，以及消息内容
            string topic_key = msg->GetTopicKey();
            string topic_msg = msg->GetTopicMsg();
            // 通过主题名称，查找对应主题的回调函数，有就处理，无在报错
            auto callback = GetSubscribe(topic_key);
            if (!callback)
            {
                ELog("收到了%s 主题消息，但是该消息无法处理回调", topic_key.c_str());
                return;
            }
            return callback(topic_key, topic_msg);
        }

    private:
        void AddSubscribe(const string &key, const SubCallback &cb)
        {
            unique_lock<mutex> lock(_mutex);
            _topic_call_back.insert(make_pair(key, cb));
        }
        void DelSubscribe(const string &key)
        {
            unique_lock<mutex> lock(_mutex);
            _topic_call_back.erase(key);
        }
        const SubCallback GetSubscribe(const string &key)
        {
            unique_lock<mutex> lock(_mutex);
            auto it = _topic_call_back.find(key);
            if (it == _topic_call_back.end())
            {
                return SubCallback();
            }
            return it->second;
        }
        bool CommRequest(const BaseConnection::ptr &conn, const std::string &key,
                         TopicOptype type, const std::string &msg = "")
        {
            //构造请求对象，填充数据
            auto msg_req = MessageFactory::create<TopicRequest>();
            msg_req->SetId(UUID::uuid());
            msg_req->SetMtype(Mtype::REQ_TOPIC);
            msg_req->SetOptype(type);
            msg_req->SetTopicKey(key);
            if (type == TopicOptype::TOPIC_PUBLISH)
            {
                msg_req->SetTopicMsg(msg);
            }
            // 2. 向服务端发送请求，等待响应
            BaseMessage::ptr msg_rsp;
            bool ret = _requestor->Send(conn, msg_req, msg_rsp);
            if (ret == false)
            {
                ELog("主题操作请求失败！");
                return false;
            }
            // 3. 判断请求处理是否成功
            
            auto topic_rsp_msg = std::dynamic_pointer_cast<TopicResponse>(msg_rsp);
            if (!topic_rsp_msg)
            {
                ELog("主题操作响应，向下类型转换失败！");
                return false;
            }
            if (topic_rsp_msg->GetRcode() != RCode::RCODE_OK)
            {
                printf("code: %d\n", topic_rsp_msg->GetRcode());
                ELog("主题操作请求出错：%s", ErrReason(topic_rsp_msg->GetRcode()).c_str());
                return false;
            }
            return true;
        }

    private:
        mutex _mutex;
        unordered_map<string, SubCallback> _topic_call_back;
        Requestor::ptr _requestor;
    };
}
